package org.flink.marketing.jobs;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
//import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.flink.marketing.sink.ClickHouseSink;
import org.flink.marketing.sink.ClickHouseTestSink;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

@Slf4j
public class EventTrackingJob {
    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try (InputStream inputStream = EventTrackingJob.class.getResourceAsStream("/kafka.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        log.info("EventTrackingJob flink job 启动");
        init();
        log.info("EventTrackingJob flink job 加载结束");
        //设置环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        String threadCountEnv = System.getenv("THREAD_COUNT");
        int threadCount = 4;
        if (threadCountEnv != null) {
            threadCount = Integer.parseInt(threadCountEnv);
        }
        //获取数据-数据源
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(properties.getProperty("bootstrap.servers"))
                .setTopics("maidian-data")
                .setGroupId(properties.getProperty("group.id"))
//                .setStartingOffsets(OffsetsInitializer.earliest())
                .setStartingOffsets(OffsetsInitializer.latest())
                .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
                // 启用自动提交偏移量，并设置提交间隔（单位：毫秒）
                .setProperty("enable.auto.commit", "true")
                .setProperty("auto.commit.interval.ms", "1000") // 每秒提交一次偏移量
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
//
        DataStreamSource<String> trackings = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source").setParallelism(threadCount);
//        DataStreamSource<String> trackings = env.addSource(new FlinkKafkaConsumer("maidian-data", new SimpleStringSchema(), properties));
//        DataStreamSource<String> trackings = env.addSource(new FlinkKafkaConsumer("maidian-data", new SimpleStringSchema(), properties));;
        //trackings.setParallelism(4);
//        trackings.name("从kafka读取上行埋点数据");
//        trackings.print();

        try{
            DataStream<JSONObject> trackResult = trackings.flatMap(new FlatMapFunction<String, JSONObject>() {
                public void flatMap(String value, Collector<JSONObject> out){
                    JSONObject jsonObject = new JSONObject();
                    try {
                        log.info("EventTrackingJob 获取kafka数据:"+value);
                        jsonObject = JSON.parseObject(value);
                    }catch (Exception e){
                        log.info("EventTrackingJob 解析数据异常:"+value);
                        e.printStackTrace();
                    }

                    out.collect(jsonObject);
                }
            }).name("maidian JSONString转换为 JSONOBJECT");

            trackResult.print().name("");

            trackResult.addSink(new ClickHouseSink()).setParallelism(4).name("flink-ck");
        }catch (Exception e){
            e.printStackTrace();
            log.info("EventTrackingJob 解析数据异常");
        }




        env.execute("读取kafka，数据过滤");
    }
}
